/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Auzdora on 2023/09/07.
//

#include "common/log/log.h"
#include "rc.h"
#include "sql/expr/tuple.h"
#include "sql/expr/tuple_cell.h"
#include "sql/operator/aggregation_operator.h"
#include "sql/parser/parse_defs.h"
#include "storage/record/record.h"
#include "storage/common/table.h"

RC AggregationOperator::open()
{
  if (children_.size() != 1) {
    LOG_WARN("predicate operator must has one child");
    return RC::INTERNAL;
  }

  return children_[0]->open();
}

RC AggregationOperator::next()
{
  // should be called only once
  RC rc = RC::SUCCESS;
  bool init = true;
  int cnt = 0;
  int avg_idx = -1;

  // first compute the whole aggregation, because aggregation is a pipeline breaker
  while ((rc = children_[0]->next()) == RC::SUCCESS) {
    // get current record
    TupleCell tuple_cell;
    Tuple * tuple = children_[0]->current_tuple();
    if (nullptr == tuple) {
      rc = RC::INTERNAL;
      LOG_WARN("failed to get current record. rc=%s", strrc(rc));
      break;
    }
    for (int i = 0; i < tuple->cell_num(); i++) {
        if (aggr_infos_[i].aggr_type == AggrType::COUNTS_STAR) {
            do_aggr(i, init, NULL);
            break;
        }
        if (avg_idx == -1 && aggr_infos_[i].aggr_type == AggrType::AVGS) {
            avg_idx = i;
        }
        rc = tuple->cell_at(i, tuple_cell);
        if (rc != RC::SUCCESS) {
            LOG_WARN("failed to fetch field of cell. index=%d, rc=%s", i, strrc(rc));
            break;
        }
        do_aggr(i, init, tuple_cell);
    }

    if (init) {
        init = false;
    }
    cnt++;

  }

  // do avg
  if (avg_idx != -1) {
    aggr_results_[avg_idx] = aggr_results_[avg_idx].tuple_div(&cnt, INTS);
  }

  return RC ::SUCCESS;
}

void AggregationOperator::do_aggr(int idx, bool init, const TupleCell &cell) {
  if (init) {
    switch (aggr_infos_[idx].aggr_type) {
        case MINS:
            aggr_results_.push_back(cell);
            break;
        case MAXS:
            aggr_results_.push_back(cell);
            break;
        case AVGS:
            aggr_results_.push_back(cell);
            break;
        case COUNTS:
            aggr_results_.push_back(TupleCell(1));
            break;
        case COUNTS_STAR:
            aggr_results_.push_back(TupleCell(1));
            break;
        case INVALID:
            break;
    }
  } else {
    switch (aggr_infos_[idx].aggr_type) {
        case MINS:
            do_min(idx, cell);
            break;
        case MAXS:
            do_max(idx, cell);
            break;
        case AVGS:
            do_avg(idx, cell);
            break;
        case COUNTS:
            do_count(idx);
            break;
        case COUNTS_STAR:
            do_count(idx);
            break;
        case INVALID:
            break;
    }
  }
}

RC AggregationOperator::do_min(int idx, const TupleCell &cell) {
    int ans = aggr_results_[idx].compare(cell);
    if (ans > 0) {
        aggr_results_[idx] = cell;
        return RC::SUCCESS;
    }
    return RC::SUCCESS;
}

RC AggregationOperator::do_max(int idx, const TupleCell &cell) {
    int ans = aggr_results_[idx].compare(cell);
    if (ans < 0) {
        aggr_results_[idx] = cell;
        return RC::SUCCESS;
    }
    return RC::SUCCESS;
}

RC AggregationOperator::do_avg(int idx, const TupleCell &cell) {
    aggr_results_[idx] = aggr_results_[idx].tuple_add(cell);
    return RC::SUCCESS;
}

RC AggregationOperator::do_count(int idx) {
    aggr_results_[idx].inc_data_cnt();
    return RC::SUCCESS;
}

RC AggregationOperator::close()
{
  children_[0]->close();
  return RC::SUCCESS;
}

Tuple *AggregationOperator::current_tuple()
{
  return children_[0]->current_tuple();
}

RC AggregationOperator::tuple_cell_spec_at(int index, const TupleCellSpec *&spec) const
{
  return children_[0]->current_tuple()->cell_spec_at(index, spec);
}

const TupleCell &AggregationOperator::get_tuple_cell_as(int idx) {
    return aggr_results_[idx];
}

const AggrType AggregationOperator::get_type_at(int idx) {
    return aggr_infos_[idx].aggr_type;
}